package com.bj58.groupbuy.action.util;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;


import com.bj58.dsap.api.scf.contract.IHbaseApi;
import com.bj58.dsap.api.scf.contract.Result;

/**
 * Hbase工具类
 * @author houzuoxin
 * */
public class HbaseUtil {
	private static IHbaseApi hbase;
	static {
		ScfUtil.initSCF();
		hbase=ScfUtil.getHbaseService();
	}
	
	   public static String RowKeyToString(TreeMap<String, String> rowKeyMap){
			StringBuffer sRowKey=new StringBuffer();
			 @SuppressWarnings("rawtypes")
			Iterator it = rowKeyMap.values().iterator();
			  while(it.hasNext()){
				  Object Stmp=it.next();
				  if(Stmp!=null)
				  sRowKey.append("_"+String.valueOf(Stmp));
			  }
			return sRowKey==null?null:String.valueOf(sRowKey).substring(1);
		}
	  
	   /*
	   * getDataByKeyPrefix 通过前缀匹配查询
	   * @param tableName   表名
	   * @param rowkey      键值前缀
	   */
	  	public static Map<String, Map<String, Object>> getDataByKeyPrefix(String tableName, String rowkey) {
	  		String cacheKey = tableName + "_" + rowkey;
	  		@SuppressWarnings("unchecked")
	  		Map<String, Map<String, Object>> mapDataValue = (HashMap<String, Map<String, Object>>) CacheProvider.getCache().get(cacheKey);
	  		if (mapDataValue == null) {
	  			List<Result> listResult = null;
	  			mapDataValue = new HashMap<String, Map<String, Object>>();
	  			Boolean isNext = true;
	  			String strNextKey = "";
	  			int intPageSize = 20;
	  			while (isNext) {
	  				listResult =hbase.nextPagesByKeyPrefix(tableName, rowkey, strNextKey,intPageSize); //DSAPProvider.Instance().hbaseApi.nextPagesByKeyPrefix(tableName, rowkey, strNextKey,intPageSize);
	  				for (Result result : listResult)
	  				{
	  					if(result.getScfMap().get("rowKey")!=null)
	  					{
	  					 mapDataValue.put(result.getScfMap().get("rowKey").toString(), result.getScfMap());
	  					}
	  				}
	  				if (listResult.size() == intPageSize)
	  					strNextKey = listResult.get(intPageSize - 1).getScfMap().get("rowKey").toString();
	  				else
	  					isNext = false;
	  			}
	  		}
	  		return mapDataValue;
	  	}
	  	/* getDataByRowkey  通过表名和键值 获取
	  	 * @param tableName 表名
	  	 * @param rowkey    键值 
	  	 */
	  		public static Map<String, Object> getDataByRowkey(String tableName,String rowkey) {
	  			//System.out.println(tableName+":"+rowkey);
	  			String cacheKey = tableName + "_" + rowkey;
	  			@SuppressWarnings("unchecked")
	  			Map<String, Object> rowValue = (Map<String, Object>) CacheProvider.getCache().get(cacheKey);
	  			if (rowValue == null) {
	  				// load data from HBase
	  				Result rs = null;
	  				try {
	  					rs = hbase.queryByRowKey(tableName,rowkey);
	  					rowValue = rs.getScfMap();
	  				} catch (Exception e) {
	  				}
	  			}
	  			return rowValue;
	  		}
	  		/**
	  		 * 前缀匹配返回接下来的100条，可以设置查询起点，方便下次接着调用时接着查询。
	  		 * @param tableName 表名
	  		 * @param rowkeyPrefix    匹配的前缀
	  		 * @param nextKey 查询rowkey起点: ""为从头开始，null为查完了
	  		 * @return  Map<String nextKey, List<Map<String, Object>> realResultList> 为null表示没有查的必要了
	  		 * 
	  		 */
	  		public static Map<String, List<Map<String, Object>> > getNext100ByKeyPrefix(String tableName, String rowkeyPrefix, String nextKey) 
	  		{
	  			System.out.println(tableName+":"+rowkeyPrefix);
	  			//如果已经查完，nextKey=null，返回null表示查不到结果了
	  			if(nextKey == null || nextKey.equals("null"))
	  			{
	  				return null;
	  			}
	  			
	  			//最终返回值
	  			Map<String, List<Map<String, Object>>> mapDataValue = new HashMap<String, List<Map<String, Object>> >() ;
	  			//每次HBaseAPI查询返回值
	  			List<Result> listResult = null;
	  			List<Map<String, Object>> realResultList = new ArrayList<Map<String,Object>>();
	  			String strNextKey = nextKey;
	  			int intPageSize = 100;
	  			
	  			//实际调用HBaseAPI返回100条记录
	  			listResult =hbase.nextPagesByKeyPrefix(tableName, rowkeyPrefix, strNextKey,intPageSize);
	  			//System.out.println(listResult);
	  			for (Result result : listResult)
	  			{
	  				realResultList.add(result.getScfMap());
	  			}
	  			if (listResult.size() == intPageSize&&listResult.get(intPageSize - 1).getScfMap().get("rowKey")!=null)
	  			{		
	  				 strNextKey = listResult.get(intPageSize - 1).getScfMap().get("rowKey").toString();
	  			}
	  			else
	  			{
	  				//如果已经查完，nextKey=null
	  				strNextKey = null;
	  			}
	  			mapDataValue.put(strNextKey, realResultList);
	  			if(realResultList == null || realResultList.isEmpty())
	  			{
	  				return null;
	  			}
	  			return mapDataValue;
	  		}
	
}
